package com.zhangdi.mgsync.core.thread;

import com.mongodb.client.MongoCollection;
import com.zhangdi.mgsync.core.cache.Cache;
import com.zhangdi.mgsync.core.rabbit.AdminTemplate;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.QueueInformation;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangdi
 * @version 1.0
 * @date 2021.11.9 11:21
 */
@Component
public class ThreadManagement {
    Logger logger = LoggerFactory.getLogger(ThreadManagement.class);

    private final Map<String, SyncWorker> workers = new HashMap<>();
    private final MonitorThreadPool threadPoolExecutor;

    private final long task_batch_timeout = 10 * 1000;
    private final int task_stop_duration = 1 * 6 * 1000;//停止监听间隔时间
    private final int task_stop_delay = 1 * 6 * 1000;
    private final int task_restart_duration = 10 * 60 * 1000;
    private final int task_restart_delay = 10 * 60 * 1000;

    public ThreadManagement() {
        MineThreadFactory factory = new MineThreadFactory();
        factory.setThreadNamePrefix("MGSyncWatcher");

        threadPoolExecutor = new MonitorThreadPool(
                16, 20, 40,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(16),
                factory,
                new ThreadPoolExecutor.AbortPolicy());
    }

    public void submit(AdminTemplate adminTemplate, MongoCollection<Document> collection, Cache<String, String> cache, int second) {
        String collName = collection.getNamespace().getCollectionName();
        if (isRunning(collName)) {
            logger.info("{} is running", collName);
            return;
        }
        logger.info("启动对{}的监听.", collName);

        SyncWorker worker = new WatchThread(adminTemplate, collection, cache, second);

        //批量发送任务
        BatchSendTimeoutScheduleder.get().scheduleWithFixedDelay(
                worker::batchPostTimeoutTasker,
                task_batch_timeout,
                task_batch_timeout);

        //限流任务
        BatchSendTimeoutScheduleder.get().scheduleWithFixedDelay(
                worker::stopListener, task_stop_delay, task_stop_duration);

        //恢复任务
        BatchSendTimeoutScheduleder.get().scheduleAtFixedRate(
                () -> {
                    if (worker.tryReStartListener()) {
                        logger.info("{} 重新监听...", collName);
                        //启动
                        if(worker.isShutdown()) {
                            threadPoolExecutor.execute(worker);
                        } else {
                            worker.startAtTime();
                        }
                    }
                }, task_restart_delay, task_restart_duration);

        threadPoolExecutor.execute(worker);
        cache(worker);
    }

    public void shutdown() {
        Iterator<Map.Entry<String, SyncWorker>> iterator = workers.entrySet().iterator();
        while (iterator.hasNext()) {
            SyncWorker worker = iterator.next().getValue();
            worker.shutdown();
            threadPoolExecutor.remove(worker);
        }
    }

    public void shutdown(String queue) {
        if (!workers.containsKey(queue)) {
            return;
        }
        SyncWorker worker = workers.get(queue);
        worker.shutdown();
        threadPoolExecutor.remove(worker);
    }

    protected void cache(SyncWorker worker) {
        workers.put(worker.name(), worker);
    }

    protected boolean isRunning(String name) {
        if (!workers.containsKey(name)) {
            return false;
        }

        SyncWorker worker = workers.get(name);

        return worker.isRunning();
    }
}
